package club.hicode.daydayup.zookeeper.distribute0;

import cn.hutool.core.util.StrUtil;
import lombok.Data;
import org.apache.zookeeper.*;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.stream.Collectors;

/**
 * ZK的分布式锁
 *
 * @author Liu Chunfu
 * @date 2018-06-07 下午2:01
 **/
@Data
public class ZKLock implements java.util.concurrent.locks.Lock, Watcher {

    /**
     * 锁定的根路径
     */
    private static final String LOCK = "/lock";

    private static final byte[] DEFAULT_BYTE = "0".getBytes();

    /**
     * 链接字符串
     */
    private static final String CONNECT_STR = "39.108.231.92:2181";

    private CountDownLatch connectCountDownLatch = new CountDownLatch(1);

    private CountDownLatch countDownLatch = new CountDownLatch(1);

    private String curPath;


    private ZooKeeper zooKeeper;

    public ZKLock() {
        try {
            zooKeeper = new ZooKeeper(CONNECT_STR, 10 * 10 * 1000, this::process);
            connectCountDownLatch.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void lock() {
        try {
            String curPath = zooKeeper.create(LOCK + "/lock-", DEFAULT_BYTE, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            this.curPath = curPath;
            List<String> children = zooKeeper.getChildren(LOCK, false);
            List<String> collect = children.stream().sorted(String::compareTo).collect(Collectors.toList());
            //获取了所
            if (collect.size() == 0) {
                return;
            }
            String smallest = LOCK + "/" + collect.get(0);
            if (smallest.equals(curPath)) {
                System.out.println(curPath + " 获取了锁,任务中。。。");
            } else {
                String solved = StrUtil.removePrefix(curPath, LOCK + "/");

                List<String> collect1 = children.stream().sorted().filter(v -> solved.compareTo(v) > 0).collect(Collectors.toList());
                String prePath = LOCK + "/" + collect1.get(collect1.size() - 1);
                //没有获取锁
                zooKeeper.exists(prePath, true);

                countDownLatch.await();
                System.out.println(curPath + " 获取了锁,经过竞争。");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                zooKeeper.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        try {
            System.out.println("删除节点：" + curPath);
            zooKeeper.delete(curPath, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
            System.err.println("当前节点：" + curPath);

        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    /**
     * Zookeeper 连接后触发的事件
     *
     * @param event
     */
    @Override
    public void process(WatchedEvent event) {
        Event.EventType type = event.getType();
        Event.KeeperState state = event.getState();

        if (state == Event.KeeperState.SyncConnected) {
            connectCountDownLatch.countDown();
        }

        if (type == Event.EventType.NodeDeleted) {
            countDownLatch.countDown();
        }

    }


}
